package com.xxyuan.batch._37_step_part;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.Resource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

import java.util.List;

@SpringBootApplication
@EnableBatchProcessing
public class PartStepJob {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    @StepScope
    public FlatFileItemReader<User> itemReader(@Value("#{stepExecutionContext['file']}") Resource resource){
        return new FlatFileItemReaderBuilder<User>()
                .name("userItemReader")
                .resource(resource)
                .delimited().delimiter("#")
                .names("id","name","age")
                .targetType(User.class)
                .build();
    }

    @Bean
    public ItemWriter<User> itemWriter(){
        return new ItemWriter<User>() {
            @Override
            public void write(List<? extends User> items) throws Exception {
                items.forEach(System.err::println);
            }
        };
    }

    @Bean
    public Step workStep(){
        return stepBuilderFactory
                .get("workStep")
                .<User,User>chunk(10)
                .reader(itemReader(null))
                .writer(itemWriter())
                .build();
    }

    /**
     * 定义分区器
     */
    @Bean
    public UserPartitioner partitioner(){
        return new UserPartitioner();
    }

    /**
     * 分区处理器定义
     */
    @Bean
    public PartitionHandler partitionHandler(){
        TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
        //分区步骤个数
        handler.setGridSize(5);
        //一个从步骤是一个独立线程
        handler.setTaskExecutor(new SimpleAsyncTaskExecutor());
        //关联从步骤
        handler.setStep(workStep());

        try {
            handler.afterPropertiesSet();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return handler;
    }

    /**
     * 主步骤
     */
    @Bean
    public Step masterStep(){
        return stepBuilderFactory
                .get("masterStep")
                .partitioner(workStep().getName(),partitioner())
                .partitionHandler(partitionHandler())
                .build();
    }

    @Bean
    public Job job(){
        return jobBuilderFactory
                .get("part-step-job")
                .start(masterStep())
                .build();
    }

    public static void main(String[] args) {
        SpringApplication.run(PartStepJob.class,args);
    }
}
